package ithink.link.base.zk;

/**
 * Created by zaqb on 17-9-10.
 */
import io.netty.util.internal.ConcurrentSet;
import ithink.link.base.service.IConfigMonitor;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.zookeeper.Watcher;
import org.simpleframework.xml.Serializer;
import org.simpleframework.xml.core.Persister;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;


class ZkProxy<T> implements IZkChildListener, IZkDataListener,IZkStateListener, IConfigMonitor<T> {
    private static final Logger LOGGER = LoggerFactory.getLogger( ZkProxy.class);

    private ConcurrentHashMap<String, T> keyValues = new ConcurrentHashMap<String, T>();
    private ConcurrentSet<T> nowValues = new ConcurrentSet<T>();
    private ConcurrentSet<String> monitorPaths = new ConcurrentSet<String>();
    private ArrayList<T> newValues = new ArrayList<T>();
    private ArrayList<T> deleteValues = new ArrayList<T>();
    private ArrayList<T> updateValues = new ArrayList<T>();
    private IListener<T> listener;
    private ZkClient client;
    private String rootPath;
    private boolean started = false;
    private Class<T> type;


    public ZkProxy(Class<T> type, ZkClient client, String path, IListener<T> listener) {
        this.listener = listener;
        this.client = client;
        this.rootPath = path;
        this.type = type;
        this.client.subscribeStateChanges(this);
    }

    @Override
    public synchronized void createPersistentConfig(String path, T object) throws Exception {
        String fullPath = getFullPath(path);
        LOGGER.info("创建持久节点 path {} with object {}", fullPath, object.getClass().getName());
        try {
            byte[] bytes = toByte(object);
            if (bytes == null) {
                throw new Exception("obj to byte[] failed");
            }
            createNode(fullPath);
            client.createPersistent(fullPath, bytes);
        } catch (Exception ex) {
            LOGGER.error("set service error rootPath {}", ex);
            throw ex;
        }
    }

    @Override
    public synchronized void createTemporaryConfig(String path, T object) throws Exception {
        String fullPath = getFullPath(path);
        LOGGER.info("创建临时节点 path {} with object {}", fullPath, object.getClass().getName());
        try {
            byte[] bytes = toByte(object);
            if (bytes == null) {
                throw new Exception("obj to byte[] failed");
            }
            client.createEphemeral(fullPath, bytes);
        } catch (Exception ex) {
            LOGGER.error("set service error rootPath {}", ex);
            throw ex;
        }
    }

    @Override
    public synchronized void deleteConfig(String path) {
        String fullPath = getFullPath(path);
        LOGGER.info("删除节点 path " + fullPath);
        client.delete(fullPath);
    }

    @Override
    public synchronized T getConfig(String path) throws Exception {
        String fullPath = getFullPath(path);
        LOGGER.debug("获取节点数据 path " + fullPath);
        byte[] bytes = client.readData(fullPath, true);
        if (bytes == null) {
            throw new Exception("failed to get object bytes == null,from path " + path);
        }
        T obj = fromByte(bytes);
        if (obj == null) {
            throw new Exception("failed to get object tobytes == null,from rootPath" + path);
        }
        return obj;
    }

    public synchronized List<T> getValues() {
        return new ArrayList<>(nowValues);
    }

    @Override
    public void setConfig(String path, T config) throws Exception {
        String fullPath = getFullPath(path);
        byte[] bytes = toByte(config);
        if (bytes == null) {
            throw new Exception("obj to byte[] failed");
        }
        client.writeData(fullPath, bytes);
    }

    public void setListener(IListener<T> listener) {
        this.listener = listener;
    }

    @Override
    public synchronized void start() {
        if (!started) {
            LOGGER.info("start zk proxy on path {} with object {} ",rootPath,type.getName());
            pullAll();
            fireListener();
            started = true;
        }
    }

    @Override
    public synchronized void stop() {
        if (started) {
            client.unsubscribeChildChanges(rootPath, this);
            for (String child : monitorPaths) {
                client.unsubscribeDataChanges(child, this);
            }
            nowValues.clear();
            keyValues.clear();
            newValues.clear();
            updateValues.clear();
            deleteValues.clear();
            monitorPaths.clear();
            client.close();
            started = false;
        }
    }

    @Override
    public boolean tryCreateTemporaryPath(String path, T object) throws Exception {
        String fullPath = getFullPath(path);
        LOGGER.info("尝试创建临时节点 path {} with object {}", fullPath, object.getClass().getName());
        try {
            byte[] bytes = toByte(object);
            if (bytes == null) {
                throw new Exception("obj to byte[] failed");
            }
            client.createEphemeral(fullPath, bytes);
        } catch (ZkNodeExistsException ex) {
            return false;
        } catch (Exception ex) {
            LOGGER.error("创建临时节点失败", ex);
            throw ex;
        }
        return true;
    }

    @Override
    public synchronized void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {

        LOGGER.debug("子节变化 parent path {},children {}", parentPath, currentChilds);
        checkPath(parentPath);
        if (currentChilds != null) {
            for (String child : currentChilds) {
                checkPath(parentPath + "/" + child);
            }
        }
        fireListener();

    }

    @Override
    public synchronized void handleDataChange(String dataPath, Object data) throws Exception {
        LOGGER.debug("节点数据变化 path {}", dataPath);
        byte[] bytes = (byte[]) data;
        if (keyValues.containsKey(dataPath)) {
            T old = keyValues.get(dataPath);
            nowValues.remove(old);
            keyValues.remove(dataPath);
            try {
                if (bytes != null) {
                    LOGGER.debug("节点 {},更新数据\n {}", dataPath, new String((byte[]) data));
                    T updated = fromByte(bytes);
                    nowValues.add(updated);
                    keyValues.put(dataPath, updated);
                    updateValues.add(updated);
                } else {
                    LOGGER.debug("节点 {},更新数据 null", dataPath);
                    updateValues.add(null);
                }

            } catch (Exception ex) {
                ex.printStackTrace();
                LOGGER.error("parse update data error", ex);
            }
        } else {
            try {
                if (bytes == null || bytes.length == 0) {
                    return;
                }
                T newObj = fromByte(bytes);
                LOGGER.debug("节点 {},创建数据\n {}", dataPath, new String((byte[]) data));
                if (newObj != null) {
                    keyValues.put(dataPath, newObj);
                    nowValues.add(newObj);
                    newValues.add(newObj);
                }
            }
            catch (Exception ex)
            {
                ex.printStackTrace();
                LOGGER.error("parse new data error", ex);
            }
        }
        fireListener();
    }

    @Override
    public synchronized void handleDataDeleted(String dataPath) throws Exception {
        LOGGER.debug("节点数据删除 path {}", dataPath);
        if (keyValues.containsKey(dataPath)) {
            T obj = keyValues.get(dataPath);
            nowValues.remove(obj);
            keyValues.remove(dataPath);
            deleteValues.add(obj);
        } else {
            deleteValues.add(null);
        }
        fireListener();
    }

    @Override
    public synchronized void handleStateChanged(Watcher.Event.KeeperState state) throws Exception {
        LOGGER.info("zk client state to {}", state);
        if (state == Watcher.Event.KeeperState.Expired) {
            listener.onSessionExpired();
        }
    }

    @Override
    public synchronized void handleNewSession() throws Exception {
        LOGGER.info("new zk session create");
        listener.onNewSession();
    }



    public synchronized void pullAll() {
        synchronized (this) {
            checkPath(rootPath);
        }
    }

    private void checkPath(String path) {
        if (path == null || path.length() == 0) {
            return;
        }
        LOGGER.debug("遍历节点 {}", path);
        if (!monitorPaths.contains(path)) {
            try{
                client.subscribeChildChanges(path, this);
                LOGGER.debug("订阅子节点变化 Path {}", path);
                monitorPaths.add(path);
                LOGGER.debug("订阅数据变化 Path {}", path);
                client.subscribeDataChanges(path, this);
                byte[] data = client.readData(path, true);
                if (data != null && data.length > 0) {
                    T obj = fromByte(data);
                    LOGGER.debug("节点 {} 数据 \n {}", path, new String(data));
                    if (obj != null) {
                        if (!keyValues.containsKey(path)) {
                            keyValues.put(path, obj);
                            nowValues.add(obj);
                            newValues.add(obj);
                        }
                    } else {
                        LOGGER.error("get data from {} get null", path);
                    }
                }
                List<String> children = client.getChildren(path);
                for (String child : children) {
                    checkPath(path + "/" + child);
                }
            }
            catch (Exception ex)
            {
                LOGGER.error("check path {} error",path,ex);
            }
        }
    }

    private void createNode(String path) {
        path = path.replace("\\", "/");
        path = path.replace("//", "/");
        String[] arr = path.split("/");
        String fullPath = "";
        for (int i = 0; i < arr.length - 1; i++) {
            if (!"".equals(arr[i])) {
                fullPath += "/" + arr[i];
                if (!client.exists(fullPath)) {
                    client.createPersistent(fullPath);
                }
            }
        }
    }

    private void fireListener() {
        try {
            if (listener != null) {
                if (newValues.size() > 0) {
                    listener.onObjectCreated(newValues);
                }
                if (updateValues.size() > 0) {
                    listener.onObjectUpdated(updateValues);
                }
                if (deleteValues.size() > 0) {
                    listener.onObjectDeleted(deleteValues);
                }
            }
        } catch (Exception ex) {
            LOGGER.error("proxy callback error", ex);
        }

        newValues.clear();
        updateValues.clear();
        deleteValues.clear();

    }

    private T fromByte(T obj, byte[] bytes) throws Exception {
        if (bytes == null) {
            return null;
        }
        Serializer serializer = new Persister();
        ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
        try {
            return serializer.read(obj, stream, false);
        } catch (Exception e) {
            LOGGER.error("from bytes error", e);
            throw e;
        }
    }

    private T fromByte(byte[] bytes) {
        if (bytes == null) {
            return null;
        }
        Serializer serializer = new Persister();
        ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
        try {
            return serializer.read(type, stream, false);
        } catch (Exception e) {
            LOGGER.error("from bytes error", e);
            return null;
        }
    }

    private byte[] toByte(T object) {

        if (object == null) {
            return null;
        }
        Serializer serializer = new Persister();
        ByteArrayOutputStream stream = new ByteArrayOutputStream();
        try {
            serializer.write(object, stream);
            return stream.toByteArray();
        } catch (Exception e) {
            LOGGER.error("to bytes error", e);
            return null;
        }
    }

    private String getFullPath(String path) {
        if(!rootPath.endsWith("/") && !path.startsWith( "/" )) {
            return rootPath + "/" + path;
        }
        return  rootPath + path;
    }

}
